Конкурентное исполнение (Параллелизм с передачей сообщений (Concurrency)) (??? debe revisarse)
===============================================================================================

.. index::
    Конкурентное исполнение
    Параллелизм с передачей сообщений
    Concurrency

Параллелизма с передачей сообщений похож, но отличается от темы предыдущей главы, **Параллелизм(???)**.
Поскольку обе эти концепции подразумевают выполнение программ в потоках,
и поскольку параллелизм с передачей сообщений основан на параллелизме, их иногда путают друг с друг с другом.
Различия между параллелизмом с передачей сообщений и обычным параллелизмом заключаются в следующем:
* Основная цель параллелизма - использовать преимущества ядер микропроцессора
для повышения производительности программ. С другой стороны, параллелизм - это
концепция, которая может понадобиться даже в одноядерной среде. Параллелизм с передачей сообщений
заключается в том, чтобы заставить программу работать более чем в одном потоке одновременно. На
Примером параллельной программы может быть серверная программа, которая
отвечает на запросы более чем одного клиента одновременно.
* При параллелизме задачи независимы друг от друга. Фактически, это было бы
ошибкой, если бы они зависели от результатов других задач, выполняющихся в одно и то же время.
При параллелизме с передачей сообщений нормально, что потоки зависят от результатов других потоков.
потоков.
* Хотя обе модели программирования используют потоки операционной системы, в
параллелизме потоки инкапсулируются понятием задачи. Параллелизм с передачей сообщений
использует потоки в явном виде.
* Параллелизм прост в использовании, и до тех пор, пока задачи независимы, легко создавать
создавать корректно работающие программы. Конкурентное исполнение просто только тогда, когда он
основан на передаче сообщений. Очень сложно написать корректные параллельные
программы, если они основаны на традиционной модели параллелизма, которая
предполагающей совместное использование данных на основе блокировок.
D поддерживает обе модели параллелизма: передачу сообщений и совместное использование данных. Мы
рассмотрим передачу сообщений в этой главе и совместное использование данных в следующей главе.


Основные Понятия
----------------
Поток: Операционные системы выполняют программы как рабочие единицы, называемые потоками. D
программы начинают выполняться начиная с функции main() в потоке, который был назначен этой
программе операционной системой. Все операции программы
обычно выполняются в этом потоке. Программа может свободно запускать другие потоки, чтобы иметь
возможность работать над несколькими задачами одновременно. Фактически, задачи, которые были
рассмотрены в предыдущей главе, основаны на потоках, которые запускаются
автоматически с помощью модуля std.parallelism.
Операционная система может приостанавливать потоки в непредсказуемое время на
непредсказуемая время. В результате даже такие простые операции, как увеличение
переменной, могут быть приостановлены в середине операции:

.. code-block:: d

    ++i;


Описанная выше операция включает в себя три шага: считывание значения переменной,
увеличение значения и присвоение нового значения обратно переменной.
Поток может быть приостановлен в любой момент между этими шагами, чтобы быть продолженным через
непредсказуемое время.
Сообщение: Данные, которые передаются между потоками, называются сообщениями. Сообщения
могут состоять из любого типа и любого количества переменных.
Идентификатор потока: У каждого потока есть идентификатор, который используется для указания
получатели сообщений.
Владелец: Любой поток, который запускает другой поток, называется владельцем нового
потока.
Рабочий поток: Любой поток, который запускается владельцем, называется рабочим (worker)

Запуск потоков
--------------

Функция spawn() принимает в качестве параметра указатель функции и запускает новый поток из
этой функции. Все операции, выполняемые этой функцией, включая
другие функции, которые она может вызывать, будут выполняться в новом потоке. Основное
разница между потоком, запущенным с помощью spawn(), и потоком, запущенным с помощью task() (:ref:`Параллелизм<глава_84_start>`),
заключается в том, что spawn() позволяет потокам посылать друг другу сообщения.
Как только запускается новый поток, владелец и  исполнитель (worker) начинают выполняться
по отдельности, как если бы они были независимыми программами:

.. code-block:: d

    import std.stdio;
    import std.concurrency;
    import core.thread;

    void worker()
    {
        foreach (i; 0 .. 5)
        {
            Thread.sleep(500.msecs);
            writeln(i, " (worker)");
        }
    }

    void main()
    {
        spawn(&worker);
        foreach (i; 0 .. 5)
        {
            Thread.sleep(300.msecs);
            writeln(i, " (main)");
        }
        writeln("main is done.");
    }

Примеры в этой главе вызывают Thread.sleep для замедления потоков, чтобы
чтобы продемонстрировать, что они выполняются в одно и то же время. Вывод программы показывает.
что два потока, один из которых выполняет main(), а другой был запущен командой
spawn(), выполняются независимо друг от друга в одно и то же время:

.. code-block:: d

    0 (main)
    0 (worker)
    1 (main)
    2 (main)
    1 (worker)
    3 (main)
    2 (worker)
    4 (main)
    main is done.
    3 (worker)
    4 (worker)

Программа автоматически ждет, пока все потоки закончат выполнение. Мы можем
это видно из приведенного выше вывода по тому, что worker() продолжает выполняться даже
после того, как main() выходит из программы, напечатав "main is done".
Параметры, которые принимает функция thread, передаются в spawn() в качестве
второй и последующие аргументы. Два рабочих потока в следующей программе
печатают по четыре числа. Они принимают начальное число в качестве параметра функции thread
параметр:

.. code-block:: d

    import std.stdio;
    import std.concurrency;
    import core.thread;

    void worker(int firstNumber)
    {
        foreach (i; 0 .. 4)
        {
            Thread.sleep(500.msecs);
            writeln(firstNumber + i);
        }
    }

    void main()
    {
        foreach (i; 1 .. 3)
        {
            spawn(&worker, i * 10);
        }
    }

Вывод, данные потока worker (отмечены цветом ???)

.. code-block:: console

    10
    20
    11
    21
    12
    22
    13
    23

Строки вывода могут быть разными в разное время в зависимости от того, как
потоки приостанавливаются и возобновляются операционной системой.
Каждая операционная система накладывает ограничения на количество потоков, которые могут существовать
одно время. Эти ограничения могут быть установлены для каждого пользователя, для всей системы или для чего-то еще
Общая производительность системы может снизиться, если число потоков, больше чем количество ядер в системе. A
поток, который активно работает в определенный момент времени, считается связанным с процессором в этот момент
во времени. С другой стороны, некоторые потоки проводят значительную часть своего
ожидая наступления какого-либо события, например ввода данных от пользователя, данных от сетевого
например, ввод от пользователя, данные от сетевого соединения, завершение вызова Thread.sleep и т. д. Считается, что такие потоки
связаны с вводом/выводом в это время. Если большинство потоков связаны с вводом/выводом, то программа может позволить себе запустить больше потоков.
программа может позволить себе запустить больше потоков, чем количество ядер, без какого-либо
ухудшения производительности. Как и в каждом решении, касающемся
касающихся производительности программы, необходимо провести реальные измерения, чтобы быть
точно убедиться в том, что это действительно так.

Идентификаторы потоков
-----------------------

thisTid() возвращает идентификатор текущего потока. Обычно она вызывается
без скобок функции:

.. code-block:: d

    import std.stdio;
    import std.concurrency;

    void printTid(string tag)
    {
        writefln("%s: %s", tag, thisTid);
    }

    void worker()
    {
        printTid("Worker");
    }

    void main()
    {
        spawn(&worker);
        printTid("Owner ");
    }

Поскольку значение id типа tid не является программно значимым, это
Даже функция toString для этого типа не определена. Поэтому в следующей части программы
мы видим только имя типа в выводе:

.. code-block:: d

    Owner : Tid(std.concurrency.MessageBox)
    Worker: Tid(std.concurrency.MessageBox)

Возвращаемое значение spawn(), которое я игнорировал до этого момента, - это идентификатор
рабочего потока:

.. code-block:: d

    Tid myWorker = spawn(&worker);

И наоборот, владельца рабочего потока можно узнать с помощью функции ownerTid().
В общем, владелец идентифицируется по ownerTid, а рабочий - по
по возвращаемому значению функции spawn().

Передача сообщений
-------------------

send() отправляет сообщения, а receiveOnly() ожидает сообщения определенного
типа. (Существуют также функции prioritySend(), receive() и receiveTimeout(), которые
которые будут описаны ниже.)
В следующей программе владелец посылает своему работнику сообщение типа int
и ожидает сообщения от рабочего типа double. Потоки продолжают
посылать сообщения туда и обратно, пока владелец не отправит отрицательное int. Это
поток-владелец:

.. code-block:: d

    void main()
    {
        Tid worker = spawn(&workerFunc);
        foreach (value; 1 .. 5)
        {
            worker.send(value);
            double result = receiveOnly!double();
            writefln("sent: %s, received: %s", value, result);
        }
        /* Sending a negative value to the worker so that it
    * terminates. */
        worker.send(-1);
    }

main() сохраняет возвращаемое значение spawn() под именем worker и использует эту
переменную при отправке сообщений рабочему.
С другой стороны, рабочий получает нужное ему сообщение в виде int, использует это значение в вычислениях и отправляет результат в виде типа double.
это значение в вычислениях и отправляет результат в виде типа double своему владельцу:

.. code-block:: d

    void workerFunc()
    {
        int value = 0;
        while (value >= 0)
        {
            value = receiveOnly!int();
            double result = to!double(value) / 5;
            ownerTid.send(result);
        }
    }

Главный поток сообщает о сообщениях, которые он отправляет, и о сообщениях, которые он
получает:

.. code-block:: d

    sent: 1, received: 0.2
    sent: 2, received: 0.4
    sent: 3, received: 0.6
    sent: 4, received: 0.8

Можно отправить несколько значений в составе одного сообщения. Cледующее сообщение состоит из трех частей:

.. code-block:: d

    ownerTid.send(thisTid, 42, 1.5);

Значения, передаваемые как части одного сообщения, появляются в виде кортежа на
на стороне получателя. В таких случаях параметры шаблона receiveOnly() должны
соответствовать типам членов кортежа:

.. code-block:: d

    /* Wait for a message composed of Tid, int, and double. */
    auto message = receiveOnly!(Tid, int, double)();
    auto sender   = message[0]; // of type Tid
    auto integer  = message[1]; // of type int
    auto floating = message[2]; // of type double

Если типы не совпадают, будет выброшено исключение MessageMismatch:

.. code-block:: d

    import std.concurrency;

    void workerFunc()
    {
        ownerTid.send("hello"); // ← Sending string
    }

    void main()
    {
        spawn(&workerFunc);
        auto message = receiveOnly!double(); // ← Expecting double
    }

Вывод:

.. code-block:: console

    std.concurrency.MessageMismatch@std/concurrency.d(235): Unexpected message type: expected 'double', got 'immutable(char)[]

Исключения, которые может бросить рабочий поток, не могут быть перехвачены владельцем. Одно из
решение - попросить рабочий поток перехватить исключение, и отправить его в виде сообщения. Мы
рассмотрим это ниже

**Пример**

Давайте воспользуемся тем, что мы видели до сих пор, в программе моделирования.
Следующая программа моделирует независимых роботов, перемещающихся
случайным образом в двумерном пространстве. Движение каждого робота обрабатывается
отдельный поток, который при запуске получает три части информации:
* Номер (id) робота: Эта информация отправляется обратно владельцу, чтобы
чтобы идентифицировать робота, с которым связано сообщение.
* Начало: Это место, откуда робот начинает движение.
* Продолжительность между каждым шагом: Эта информация используется для определения
когда робот сделает следующий шаг.
Эта информация может храниться в следующей структуре Job struct:

.. code-block:: d

    struct Job
    {
        size_t robotId;
        Position origin;
        Duration restDuration;
    }

Функция потока, который перемещает каждого робота, посылает идентификатор робота и его
движение в поток-владелец непрерывно:

.. code-block:: d

    void robotMover(Job job)
    {
        Position from = job.origin;
        while (true)
        {
            Thread.sleep(job.restDuration);
            Position to = randomNeighbor(from);
            Movement movement = Movement(from, to);
            from = to;
            ownerTid.send(MovementMessage(job.robotId, movement));
        }
    }

Владелец просто ждет этих сообщений в безусловном цикле. Он идентифицирует
роботов по идентификаторам роботов, которые отправляются в составе сообщений. Владелец
просто печатает каждое движение:

.. code-block:: d

    while (true)
    {
        auto message = receiveOnly!MovementMessage();
        writefln("%s %s",
            robots[message.robotId], message.movement);
    }

Все сообщения в этой простой программе передаются от рабочего к владельцу.
Передача сообщений обычно включает в себя более сложные коммуникации во многих
видах программ.
Вот полная версия программы:

.. code-block:: d

    import std.stdio;
    import std.random;
    import std.string;
    import std.concurrency;
    import core.thread;

    struct Position
    {
        int line;
        int column;
        string toString()
        {
            return format("%s,%s", line, column);
        }
    }

    struct Movement
    {
        Position from;
        Position to;
        string toString()
        {
            return ((from == to)
                    ? format("%s (idle)", from) : format("%s -> %s", from, to));
        }
    }

    class Robot
    {
        string image;
        Duration restDuration;
        this(string image, Duration restDuration)
        {
            this.image = image;
            this.restDuration = restDuration;
        }

        override string toString()
        {
            return format("%s(%s)", image, restDuration);
        }
    }
    /* Returns a random position around 0,0. */
    Position randomPosition()
    {
        return Position(uniform!"[]"(-10, 10),
            uniform!"[]"(-10, 10));
    }
    /* Returns at most one step from the specified coordinate. */
    int randomStep(int current)
    {
        return current + uniform!"[]"(-1, 1);
    }
    /* Returns a neighbor of the specified Position. It may be one
    * of the neighbors at eight directions, or the specified
    * position itself. */
    Position randomNeighbor(Position position)
    {
        return Position(randomStep(position.line),
            randomStep(position.column));
    }

    struct Job
    {
        size_t robotId;
        Position origin;
        Duration restDuration;
    }

    struct MovementMessage
    {
        size_t robotId;
        Movement movement;
    }

    void robotMover(Job job)
    {
        Position from = job.origin;
        while (true)
        {
            Thread.sleep(job.restDuration);
            Position to = randomNeighbor(from);
            Movement movement = Movement(from, to);
            from = to;
            ownerTid.send(MovementMessage(job.robotId, movement));
        }
    }

    void main()
    {
        /* Robots with various restDurations. */
        Robot[] robots = [
            new Robot("A", 600.msecs),
            new Robot("B", 2000.msecs),
            new Robot("C", 5000.msecs)
        ];
        /* Start a mover thread for each robot. */
        foreach (robotId, robot; robots)
        {
            spawn(&robotMover, Job(robotId,
                    randomPosition(),
                    robot.restDuration));
        }
        /* Ready to collect information about the movements of the
    * robots. */
        while (true)
        {
            auto message = receiveOnly!MovementMessage();
            /* Print the movement of this robot. */
            writefln("%s %s",
                robots[message.robotId], message.movement);
        }
    }

Программа печатает каждое движение до тех пор, пока не завершится:

.. code-block:: console

    A(600 ms) 6,2 -> 7,3
    A(600 ms) 7,3 -> 8,3
    A(600 ms) 8,3 -> 7,3
    B(2 secs) -7,-4 -> -6,-3
    A(600 ms) 7,3 -> 6,2
    A(600 ms) 6,2 -> 7,1
    A(600 ms) 7,1 (idle)
    B(2 secs) -6,-3 (idle)
    A(600 ms) 7,1 -> 7,2
    A(600 ms) 7,2 -> 7,3
    C(5 secs) -4,-4 -> -3,-5
    A(600 ms) 7,3 -> 6,4
    ...

Эта программа демонстрирует, насколько полезным может быть параллелизм с передачей сообщений:
Движения роботов рассчитываются независимо друг от друга отдельными потоками без
не зная друг о друге. Именно поток-владелец сериализует процесс печати
просто получая сообщения из своего ящика сообщений одно за другим.

Ожидание сообщений разных типов
--------------------------------

receiveOnly() может ожидать только один тип сообщения. receive(), с другой стороны
может ожидать более одного типа сообщений. Она отправляет сообщения
делегатам обработки сообщений. Когда сообщение поступает, оно сравнивается с
типом сообщения каждого делегата. Делегат, соответствующий типу
конкретного сообщения, обрабатывает его.
Например, следующий вызов receive() определяет два обработчика сообщений, которые
обрабатывают сообщения типов int и string соответственно:

.. code-block:: d

    void workerFunc()
    {
        bool isDone = false;
        while (!isDone)
        {
            void intHandler(int message)
            {
                writeln("handling int message: ", message);
                if (message == -1)
                {
                    writeln("exiting");
                    isDone = true;
                }
            }

            void stringHandler(string message)
            {
                writeln("handling string message: ", message);
            }

            receive(&intHandler, &stringHandler);
        }

Сообщениям типа int соответствует intHandler(), а сообщениям типа string
будут соответствовать stringHandler(). Рабочий поток, описанный выше, можно проверить с помощью
следующей программой:

.. code-block:: d

    import std.stdio;
    import std.concurrency;

    // ...
    void main()
    {
        auto worker = spawn(&workerFunc);
        worker.send(10);
        worker.send(42);
        worker.send("hello");
        worker.send(-1); // ← to terminate the worker
    }

Вывод программы показывает, что сообщения обрабатываются соответствующими
функции на стороне получателя:

.. code-block:: console

    handling int message: 10
    handling int message: 42
    handling string message: hello
    handling int message: -1
    exiting

Лямбда-функции и объекты типов, определяющих функцию-член opCall()
также могут быть переданы в функцию receive() в качестве обработчиков сообщений. Следующий
рабочий обрабатывает сообщения с помощью лямбда-функций. Следующая программа также
определяет специальный тип Exit, который используется для сообщения потоку о том, что ему пора
что ему пора завершить работу. Использование такого специального типа более выразительно, чем отправка
произвольное значение -1, как это было сделано в предыдущем примере.
Ниже приведены три анонимные функции, которые передаются в receive() в качестве
обработчиков сообщений. Их фигурные скобки выделены:

.. code-block:: d

    import std.stdio;
    import std.concurrency;

    struct Exit
    {
    }

    void workerFunc()
    {
        bool isDone = false;
        while (!isDone)
        {
            receive(
                (int message) { writeln("int message: ", message); },
                (string message) { writeln("string message: ", message); },
                (Exit message) { writeln("exiting"); isDone = true; });
        }
    }

    void main()
    {
        auto worker = spawn(&workerFunc);
        worker.send(10);
        worker.send(42);
        worker.send("hello");
        worker.send(Exit());


**Получение сообщений любого типа**

std.variant.Variant - это тип, который может инкапсулировать любой тип данных. Сообщения
которые не соответствуют обработчикам, указанным ранее в списке аргументов
всегда соответствуют обработчику Variant:

.. code-block:: d

    import std.stdio;
    import std.concurrency;

    void workerFunc()
    {
        receive(
            (int message) { /* ... */ },
            (double message) { /* ... */ },
            (Variant message) { writeln("Unexpected message: ", message); });
    }

    struct SpecialMessage
    {
        Message Passing Concurrency 630// ...
    }
    void main()
    {
        auto worker = spawn(&workerFunc);
        worker.send(SpecialMessage());
    }

Вывод:

.. code-block:: d

    Unexpected message: SpecialMessage()


Детали Variant выходят за рамки этой главы.

Ожидание сообщений в течение определенного времени
---------------------------------------------------

Возможно, нет смысла ждать сообщений дольше определенного времени. Отправитель
может быть временно занят или завершился с исключением.
Функция receiveTimeout() предотвращает бесконечную блокировку потока получения.
Первый параметр функции receiveTimeout() определяет, как долго сообщение
следует ожидать. Возвращаемое значение - true, если сообщение было получено
в течение этого времени, а в противном случае - false.

.. code-block:: d

    import std.stdio;
    import std.concurrency;
    import core.thread;

    void workerFunc()
    {
        Thread.sleep(3.seconds);
        ownerTid.send("hello");
    }

    void main()
    {
        spawn(&workerFunc);
        writeln("Waiting for a message");
        bool received = false;
        while (!received)
        {
            received = receiveTimeout(600.msecs,
                (string message) { writeln("received: ", message); });
            if (!received)
            {
                writeln("... no message yet");
                /* ... other operations may be executed here ... */
            }
        }
    }

Владелец выше ожидает сообщения до 600 миллисекунд. Он может продолжать
работать над другими делами, если сообщение не пришло в течение этого времени:

.. code-block:: console

    Waiting for a message
    ... no message yet
    ... no message yet
    ... no message yet
    ... no message yet
    received: hello

Исключения во время выполнения рабочего потока
----------------------------------------------

Как мы видели в предыдущей главе, средства модуля std.parallelism
автоматически перехватывают исключения, возникшие во время
выполнения задач, и перевыполняют их в контексте владельца. Это позволяет
владельцу перехватывать такие исключения:

.. code-block:: d

    try
    {
        theTask.yieldForce();
    }
    catch (Exception exc)
    {
        writefln("Detected an error in the task: '%s'",
            exc.msg);
    }

std.concurrency не предоставляет такого удобства для общих типов исключений
типов. Однако исключения могут быть пойманы и отправлены явно рабочим потоком. Как
мы увидим ниже, можно также получать OwnerTerminated и
LinkTerminated в виде сообщений.
Приведенная ниже функция calculate() получает строковые сообщения, преобразует их в
double, прибавляет 0,5 и отправляет результат обратно в виде сообщения:

.. code-block:: d

    void calculate()
    {
        while (true)
        {
            auto message = receiveOnly!string();
            ownerTid.send(to!double(message) + 0.5);
        }
    }

Вызов to!double(), приведенный выше, вызовет исключение, если строка не может быть
преобразовать в двойное значение. Поскольку такое исключение приведет к завершению
рабочего потока, владелец в следующей программе может получить
ответ только на первое сообщение:

.. code-block:: d

    import std.stdio;
    import std.concurrency;
    import std.conv;

    // ...
    void main()
    {
        Tid calculator = spawn(&calculate);
        calculator.send("1.2");
        calculator.send("hello"); // ← incorrect input
        calculator.send("3.4");
        foreach (i; 0 .. 3)
        {
            auto message = receiveOnly!double();
            writefln("result %s: %s", i, message);
        }
    }

Владелец получает ответ для "1.2" как 1.7, но поскольку рабочий был
завершен, владелец будет заблокирован в ожидании сообщения, которое никогда не придет.
никогда не придет:

.. code-block:: console

    result 0: 1.7
    ← waiting for a message that will never arriv

Рабочий поток может явно перехватить исключение и отправить его
в виде специального сообщения об ошибке. Следующая программа отправляет причину сбоя
в виде сообщения CalculationFailure. Кроме того, эта программа использует преимущество
специального типа сообщения, чтобы сигнализировать рабочему о том, что пора выходить:

.. code-block:: d

    import std.stdio;
    import std.concurrency;
    import std.conv;

    struct CalculationFailure
    {
        string reason;
    }

    struct Exit
    {
    }

    void calculate()
    {
        bool isDone = false;
        while (!isDone)
        {
            receive(
                (string message) {
                try
                {
                    ownerTid.send(to!double(message) + 0.5);
                }
                catch (Exception exc)
                {
                    ownerTid.send(CalculationFailure(exc.msg));
                }
            },
                (Exit message) { isDone = true; });
        }
    }

    void main()
    {
        Tid calculator = spawn(&calculate);
        calculator.send("1.2");
        calculator.send("hello"); // ← incorrect input
        calculator.send("3.4");
        calculator.send(Exit());
        foreach (i; 0 .. 3)
        {
            writef("result %s: ", i);
            receive(
                (double message) { writeln(message); },
                (CalculationFailure message) {
                writefln("ERROR! '%s'", message.reason);
            });
        }
    }

На этот раз причина неудачи будет напечатана владельцем:

.. code-block:: console

    result 0: 1.7
    result 1: ERROR! 'no digits seen'
    result 2: 3.9

Другой метод заключается в отправке владельцу самого объекта исключения.
Владелец может использовать объект исключения или просто отбросить его

.. code-block:: d

    // ... at the worker ...
    try
    {
        // ...
    }
    catch (shared(Exception) exc)
    {
        ownerTid.send(exc);
    }},// ... at the owner ...
    receive(
        // ...
        (shared(Exception) exc)
    {
        throw exc;});

Причина, по которой необходимы общие спецификаторы, объясняется в следующей главе
главе.

Обнаружение завершения потока
------------------------------

Потоки могут обнаружить, что получатель сообщения завершился.

**Исключение OwnerTerminated**

Это исключение выбрасывается при получении сообщения от владельца, если владелец
был прерван. Промежуточный поток владельца ниже просто завершается после
отправки двух сообщений своему рабочему. Это приводит к возникновению исключения OwnerTerminated
в рабочем потоке:

.. code-block:: d

    import std.stdio;
    import std.concurrency;

    void main()
    {
        spawn(&intermediaryFunc);
    }

    void intermediaryFunc()
    {
        auto worker = spawn(&workerFunc);
        worker.send(1);
        worker.send(2);
    } // ← Terminates after sending two messages
    void workerFunc()
    {
        while (true)
        {
            auto m = receiveOnly!int(); // ← An exception is
            //   thrown if the owner
            //   has terminated.
            writeln("Message: ", m);
        }
    }

Вывод:

.. code-block:: console

    Message: 1
    Message: 2
    std.concurrency.OwnerTerminated@std/concurrency.d(248):
    Owner terminated

Рабочий может перехватить это исключение, чтобы завершить работу изящно:

.. code-block:: d

    void workerFunc()
    {
        bool isDone = false;
        while (!isDone)
        {
            try
            {
                auto m = receiveOnly!int();
                writeln("Message: ", m);
            }
            catch (OwnerTerminated exc)
            {
                writeln("The owner has terminated.");
                isDone = true;
            }
        }
    }

Вывод:

.. code-block:: console

    Message: 1
    Message: 2
    The owner has terminated.

Ниже мы увидим, что это исключение может быть получено и в виде сообщения.

**Исключение LinkTerminated**

spawnLinked() используется так же, как spawn(). Когда рабочий, который был
который был запущен с помощью spawnLinked(), завершается, исключение LinkTerminated
выбрасывается владельцу:

.. code-block:: d

    import std.stdio;
    import std.concurrency;

    void main()
    {
        auto worker = spawnLinked(&workerFunc);
        while (true)
        {
            auto m = receiveOnly!int(); // ← An exception is
            //   thrown if the worker
            //   has terminated.
            writeln("Message: ", m);
        }
    }

    void workerFunc()
    {
        ownerTid.send(10);
        ownerTid.send(20);
    } // ← Terminates after sending two messages

Приведенный выше рабочий поток завершает работу после отправки двух сообщений. Поскольку рабочий поток был
был запущен с помощью spawnLinked(), владелец получает уведомление о завершении рабочего
исключением LinkTerminated:

.. code-block:: d

    Message: 10
    Message: 20
    std.concurrency.LinkTerminated@std/concurrency.d(263):
    Link terminated

Владелец может перехватить исключение, чтобы сделать что-то особенное, например, завершить работу
изящно:

.. code-block:: d

    bool isDone = false;
    while (!isDone)
    {
        try
        {
            auto m = receiveOnly!int();
            writeln("Message: ", m);
        }
        catch (LinkTerminated exc)
        {
            writeln("The worker has terminated.");
            isDone = true;
        }
    }

Вывод:

.. code-block:: console

    Message: 10
    Message: 20
    The worker has terminated.

Это исключение может быть получено и в виде сообщения

**Получение исключений в виде сообщений**

Исключения OwnerTerminated и LinkTerminated могут быть получены в виде
сообщения. Следующий код демонстрирует это для исключения OwnerTerminated
исключения: (TODO check fomating)

.. code-block:: d

    bool isDone = false;
    while (!isDone)
    {
        receive(
            (int message)
        {
            writeln("Message: ", message);},
            (OwnerTerminated exc)
            {
                writeln("The owner has terminated; exiting."); isDone = true;
            });
    }

Управление почтовыми ящиками
----------------------------

У каждой темы есть личный почтовый ящик, в котором хранятся сообщения, отправленные в эту тему.
поток. Количество сообщений в почтовом ящике может увеличиваться или уменьшаться
в зависимости от того, сколько времени требуется потоку для получения и ответа на каждое сообщение
сообщение. Постоянно растущий почтовый ящик создает нагрузку на всю систему и
может указывать на недостатки проектирования программы. Это также может означать, что поток может
никогда не доберется до самых последних сообщений.
Функция setMaxMailboxSize() используется для ограничения количества сообщений, которые
почтовый ящик. Три ее параметра задают почтовый ящик, максимальное
количество сообщений, которое он может вместить, и что должно произойти, когда почтовый ящик
в таком порядке. Для последнего параметра есть четыре варианта:
* OnCrowding.block: Отправитель ждет, пока в почтовом ящике не освободится место.
* OnCrowding.ignore: Сообщение отбрасывается.
* OnCrowding.throwException: При отправке сообщения выбрасывается исключение MailboxFull
отправке сообщения.
* Указатель функции типа bool function(Tid): Вызывается указанная функция
вызывается.
Прежде чем рассмотреть пример использования функции setMaxMailboxSize(), давайте сначала заставим почтовый ящик
постоянно увеличиваться. Рабочий в следующей программе отправляет сообщения от одного к
назад, но владелец тратит некоторое время на каждое сообщение:

.. code-block:: d

    /* WARNING: Your system may become unresponsive when this
    *          program is running. */
    import std.concurrency;
    import core.thread;

    void workerFunc()
    {
        while (true)
        {
            ownerTid.send(42); // ← Produces messages continuously
        }
    }

    void main()
    {
        spawn(&workerFunc);
        while (true)
        {
            receive(
                (int message) {
                // Spends time for each message
                Thread.sleep(1.seconds);
            });
        }
    }

Поскольку потребитель работает медленнее производителя, память, которую использует программа
представленная выше, будет постоянно расти. Чтобы предотвратить это, владелец может ограничить
размер своего почтового ящика перед запуском рабочего:


.. code-block:: d

    void main()
    {
        setMaxMailboxSize(thisTid, 1000, OnCrowding.block);
        spawn(&workerFunc);
        // ...
    }

Вызов setMaxMailboxSize() выше устанавливает размер почтового ящика главного потока на 1000.
OnCrowding.block заставляет отправителя ждать, пока в почтовом ящике не освободится место.
В следующем примере используется OnCrowding.throwException, который вызывает
исключение MailboxFull при отправке сообщения в почтовый ящик, который
переполнен:

.. code-block:: d

    import std.concurrency;
    import core.thread;

    void workerFunc()
    {
        while (true)
        {
            try
            {
                ownerTid.send(42);
            }
            catch (MailboxFull exc)
            {
                /* Failed to send; will try again later. */
                Thread.sleep(1.msecs);
            }
        }
    }

    void main()
    {
        setMaxMailboxSize(thisTid, 1000, OnCrowding.throwException);
        spawn(&workerFunc);
        while (true)
        {
            receive(
                (int message) { Thread.sleep(1.seconds); });
        }
    }

Приоритетные сообщения
----------------------

Сообщения могут быть отправлены с более высоким приоритетом, чем обычные сообщения, с помощью функции
prioritySend(). Эти сообщения обрабатываются раньше других сообщений, которые
уже находятся в почтовом ящике:

.. code-block:: d

    prioritySend(ownerTid, ImportantMessage(100));

Если у приемника нет обработчика сообщений, соответствующего типу
приоритетного сообщения, то будет выброшено исключение PriorityMessageException:

.. code-block:: console

    std.concurrency.PriorityMessageException@std/concurrency.d(280): Priority message

Имена потоков
-------------

В простых программах, которые мы использовали выше, было легко передавать идентификаторы потоков
владельцев и рабочих. Передача идентификаторов потоков от потока к потоку может оказаться слишком
сложным в программах, использующих более пары потоков. Чтобы уменьшить эту
сложность, можно назначить потокам имена, которые будут глобально доступны
из любого потока.
Следующие три функции определяют интерфейс ассоциативного массива, к которому
к которому имеет доступ каждый поток:

* register(): Связывает поток с именем.
* locate(): Возвращает поток, связанный с указанным именем. Если
  нет потока, связанного с этим именем, то возвращается Tid.init.
* unregister(): Разрывает связь между указанным именем и
  потоком.

Следующая программа запускает два потока, которые находят друг друга по именам.
Эти потоки непрерывно посылают друг другу сообщения, пока не получат указание
завершить работу с помощью сообщения Exit:

.. code-block:: d


    import std.stdio;
    import std.concurrency;
    import core.thread;

    struct Exit
    {
    }

    void main()
    {
        // A thread whose partner is named "second"
        auto first = spawn(&player, "second");
        register("first", first);
        scope (exit)
            unregister("first");
        // A thread whose partner is named "first"
        auto second = spawn(&player, "first");
        register("second", second);
        scope (exit)
            unregister("second");
        Thread.sleep(2.seconds);
        prioritySend(first, Exit());
        prioritySend(second, Exit());
        // For the unregister() calls to succeed, main() must wait
        // until the workers terminate.
        thread_joinAll();
    }

    void player(string nameOfPartner)
    {
        Tid partner;
        while (partner == Tid.init)
        {
            Thread.sleep(1.msecs);
            partner = locate(nameOfPartner);
        }
        bool isDone = false;
        while (!isDone)
        {
            partner.send("hello " ~ nameOfPartner);
            receive(
                (string message) {
                writeln("Message: ", message);
                Thread.sleep(500.msecs);
            },
                (Exit message) {
                writefln("%s, I am exiting.", nameOfPartner);
                isDone = true;
            });
        }
    }

Вызов thread_joinAll(), который можно увидеть в конце main(), предназначен для того, чтобы заставить
владельца дождаться завершения всех рабочих потоков.

Вывод:

.. code-block:: console

    Message: hello second
    Message: hello first
    Message: hello second
    Message: hello first
    Message: hello first
    Message: hello second
    Message: hello first
    Message: hello second
    second, I am exiting.
    first, I am exiting.

Резюме
--------

* Когда потоки не зависят от других потоков, предпочтите параллелизм, который был
  было темой предыдущей главы. Рассматривайте параллелизм c передачей сщщбщений (concurrency) только тогда, когда
  когда потоки зависят от операций других потоков.
* Поскольку параллелизм путем совместного использования данных трудно реализовать корректно, предпочтите
  предпочтите параллелизм через передачу сообщений, о котором пойдет речь в этой главе.
* spawn() и spawnLinked() запускают потоки.
* thisTid - идентификатор текущего потока.
* ownerTid - идентификатор потока владельца текущего потока.
* send() и prioritySend() отправляют сообщения.
* receiveOnly(), receive() и receiveTimeout() ожидают сообщений.
* Variant соответствует любому типу сообщения.
* setMaxMailboxSize() ограничивает размер почтовых ящиков.
* register(), unregister() и locate() позволяют ссылаться на потоки по
  имени.
* При передаче сообщений могут возникать исключения: MessageMismatch,
  OwnerTerminated, LinkTerminated, MailboxFull и
  PriorityMessageException.
* Владелец не может автоматически перехватывать исключения, которые выбрасываются из
  рабочего потока.
